package com.gitee.watermaker


import java.util.UUID

import org.apache.commons.lang3.time.FastDateFormat
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks
import org.apache.flink.streaming.api.functions.source.{RichSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.watermark.Watermark
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import scala.collection.mutable.ArrayBuffer
import scala.util.Random

/**
  * Author itcast
  * Date 2020/1/6 14:45
  * Desc 演示使用Flink的Watermaker水印机制+EventTime事件时间解决订单网络延迟问题
  * 每隔5s统计一次每个用户的订单总额
  */
object OrderWaterMarkDemo {

  //订单样例类Order(订单ID、用户ID、订单金额、时间戳(事件时间))
  case class Order(orderId: String, userId: Int, money: Long, timestamp: Long, systems: Long)

  //定义一个时间转换工具
  val df: FastDateFormat = FastDateFormat.getInstance("HH:mm:ss")

  def main(args: Array[String]): Unit = {
    val senv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    //启动事件时间
    senv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    //设置多久给数据分配水印
    senv.getConfig.setAutoWatermarkInterval(200)
    //准备乱序的订单数据
    import org.apache.flink.api.scala._
    val orderDS: DataStream[Order] = senv.addSource(new RichSourceFunction[Order] {
      var isRunning = true

      override def run(ctx: SourceFunction.SourceContext[Order]): Unit = {
        while (isRunning) {
          val orderId: String = UUID.randomUUID().toString.replaceAll("-", "")
          val userId: Int = Random.nextInt(1)
          val money: Int = Random.nextInt(101)
          val timestamp: Long = System.currentTimeMillis() - Random.nextInt(5) * 1000 //减去一个随机的x秒时间表示该订单是该x秒之前产生的
          val systems: Long = System.currentTimeMillis() //系统时间
          val order = Order(orderId, userId, money, timestamp, systems)
          ctx.collect(order)
          Thread.sleep(1000)
        }
      }

      override def cancel(): Unit = {
        isRunning = false
      }
    })

    //使用事件时间给数据分配水印时间
    //BoundedOutOfOrdernessTimestampExtractor中是取当前窗口的最大事件时间和最大水印时间
    /*val watermaker: DataStream[Order] = orderDS.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[Order](Time.milliseconds(4000)) {
      //设置最大延迟时间3秒
      override def extractTimestamp(element: Order): Long = {
        //从element中获取事件时间
        val timestamp: Long = element.timestamp
        println(s"数据的事件时间:${df.format(timestamp)},数据被分配的水印时间:${df.format(getCurrentWatermark.getTimestamp)},当前系统时间:${df.format(System.currentTimeMillis())}")
        timestamp
      }
    })*/

    val watermaker: DataStream[Order] = orderDS.assignTimestampsAndWatermarks(new TimeLagWatermarkGenerator)

    //进行窗口计算
    //    val result: DataStream[Order] = watermaker.keyBy(_.userId).timeWindow(Time.seconds(5)).sum(2)

    //打印结果
    //    result.print()

    //===上面的代码已经完成但是我们想看到每个窗口的数据分布怎么办?===
    val result02: DataStream[String] = watermaker.keyBy(_.userId).timeWindow(Time.seconds(5)).apply(new WindowFunction[Order, String, Int, TimeWindow] {
      var eventTimeArr = ArrayBuffer[String]()
      var watermakerArr = ArrayBuffer[String]()
      var systemArr = ArrayBuffer[String]()

      override def apply(key: Int, window: TimeWindow, input: Iterable[Order], out: Collector[String]): Unit = {
        input.foreach(order => {
          val eventTime: String = df.format(order.timestamp)
//          val watermaker: String = df.format()
          val systems: String = df.format(order.systems)
          eventTimeArr.append(eventTime)
//          watermakerArr.append(watermaker)
          systemArr.append(systems)
        })
        val start: String = df.format(window.getStart)
        val end: String = df.format(window.getEnd)
//        val str = s"key:${key} ,窗口开始:${start},窗口结束:${end},属于该窗口的数据的事件时间:${eventTimeArr},\n水印时间:${watermakerArr},\n当前系统时间:${systemArr}"
        val str = s"key:${key} ,窗口开始:${start},窗口结束:${end},属于该窗口的数据的事件时间:${eventTimeArr},\n当前系统时间:${systemArr}"
        out.collect(str)
        eventTimeArr.clear()
//        watermakerArr.clear()
        systemArr.clear()
      }
    })
    result02.print()

    senv.execute()

  }


  //底层api
  class BoundedOutOfOrdernessGenerator extends AssignerWithPeriodicWatermarks[Order] {

    val maxOutOfOrderness = 3000L // 3.5 seconds

    var currentMaxTimestamp: Long = _

    override def getCurrentWatermark: Watermark = {
      // return the watermark as current highest timestamp minus the out-of-orderness bound
      new Watermark(currentMaxTimestamp - maxOutOfOrderness)
    }

    override def extractTimestamp(element: Order, previousElementTimestamp: Long): Long = {
      val timestamp = element.timestamp
      println(s"数据的事件时间:${df.format(timestamp)},数据被分配的水印时间:${df.format(getCurrentWatermark.getTimestamp)},当前系统时间:${df.format(System.currentTimeMillis())}")
      currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp)
      timestamp
    }
  }

  /**
    * This generator generates watermarks that are lagging behind processing time by a fixed amount.
    * It assumes that elements arrive in Flink after a bounded delay.
    */
  //底层api
  class TimeLagWatermarkGenerator extends AssignerWithPeriodicWatermarks[Order] {

    val maxTimeLag = 3000L // 5 seconds

    override def extractTimestamp(element: Order, previousElementTimestamp: Long): Long = {
      val timestamp: Long = element.timestamp
      println(s"数据的事件时间:${df.format(timestamp)},数据被分配的水印时间:${df.format(getCurrentWatermark.getTimestamp)},当前系统时间:${df.format(System.currentTimeMillis())}")
      timestamp
    }

    override def getCurrentWatermark(): Watermark = {
      // return the watermark as current time minus the maximum time lag
      //这里应该是事件时间-最大延迟时间
      new Watermark(System.currentTimeMillis() - maxTimeLag)
    }
  }

}
